package org.example.com.atguigu.day03;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Iterator;

public class WcTest {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("com.atguigu.day01.$01_RddCreate");
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 读文件
        JavaRDD<String> rdd1 = sc.textFile(args[0]);
        // rdd每个元素就是文件中的一行
        JavaPairRDD<String, Integer> rdd2 = rdd1.flatMapToPair(new PairFlatMapFunction<String, String, Integer>() {
            @Override
            public Iterator<Tuple2<String, Integer>> call(String s) throws Exception {
                String[] arr = s.split(" ");
                ArrayList<Tuple2<String, Integer>> arrayList = new ArrayList<>();
                for (String s1 : arr) {
                    Tuple2<String, Integer> stringIntegerTuple2 = new Tuple2<>(s1, 1);
                    arrayList.add(stringIntegerTuple2);
                }
                return arrayList.iterator();
            }
        });
        // rdd2 [(hadoop,1), (java,1), ...]
        JavaPairRDD<String, Integer> rdd3 = rdd2.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        rdd3.saveAsTextFile(args[1]);


    }
}
